package com.qiyu.canal.client;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;

import javax.annotation.Resource;
import java.net.InetSocketAddress;

import static com.qiyu.canal.config.CanalConfig.*;

/**
 * 单机版本客户端程序
 *
 * @Author idea
 * @Date created in 9:56 上午 2020/8/2
 */
public class CanalAloneClient {

    @Resource
    private CanalHandler canalHandler;

    public static void main(String[] args) throws Exception {

        CanalHandler canalHandler = new CanalHandlerImpl();
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(CANAL_ADDRESS, PORT), DESTINATION, ACCOUNT, PASSWORD);
        connector.connect();
        connector.subscribe(FILTER);
        connector.rollback();
        System.out.println("============== start canalAloneClient ============");
        while (true) {
            //尝试从master那边拉去binlog数据batchSize条记录，有多少取多少
            Message message = connector.getWithoutAck(BATCH_BINLOG_SIZE);
            long batchId = message.getId();
            int size = message.getEntries().size();
            if (batchId == -1 || size == 0) {
                Thread.sleep(1000);
            } else {
                canalHandler.dataHandle(message.getEntries());
            }
            connector.ack(batchId);
        }
    }
}
